package cn.itcast.flink.join;

import lombok.SneakyThrows;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Date;

/**
 * Flink 双流JOIN，基于window窗口实现cogroup，案例演示【滚动事件时间窗口cogroup联合分组】
 * todo: orderStream -> 订单数据流， detailStream -> 订单详情数据流
 *
 * @author lilulu
 */
public class TumblingWindowCoGroupDemo {
    public static void main(String[] args) throws Exception {
        // 1. 执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 2. 数据源-source
        // 2-1. order订单数据流 -> 9999
        DataStreamSource<String> orderStreamSource = env.socketTextStream("node1", 9999);
        // 2-2. detail 订单详情数据流 -> 8888
        DataStreamSource<String> orderDetailStreamSource = env.socketTextStream("node1", 8888);
        // 3. 数据转换-transformation
        SingleOutputStreamOperator<MainOrder> orderStream = orderStreamSource.filter(line -> line.trim().split(",").length == 5)
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                            private FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");

                            @SneakyThrows
                            @Override
                            public long extractTimestamp(String line, long recordTimestamp) {
                                System.out.println("order-> " + line);
                                Date orderDate = format.parse(line.split(",")[0]);
                                return orderDate.getTime();
                            }
                        })
                ).map(new MapFunction<String, MainOrder>() {
                    @Override
                    public MainOrder map(String line) throws Exception {
                        // 订单数据：2022-04-05 06:00:12,order_103,user_3,shanghai-changtai,45.00
                        String[] array = line.split(",");
                        MainOrder mainOrder = new MainOrder();
                        mainOrder.setOrderTime(array[0]);
                        mainOrder.setOrderId(array[1]);
                        mainOrder.setUserId(array[2]);
                        mainOrder.setAddress(array[3]);
                        mainOrder.setOrderMoney(Double.parseDouble(array[4]));
                        return mainOrder;

                    }
                });

        SingleOutputStreamOperator<DetailOrder> orderDetailStream = orderDetailStreamSource.filter(line -> line.trim().split(",").length == 6)
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<String>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(new SerializableTimestampAssigner<String>() {
                            private FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");

                            @SneakyThrows
                            @Override
                            public long extractTimestamp(String value, long recordTimestamp) {
                                System.out.println("detail -> " + value);
                                String[] split = value.split(",");
                                Date orderDetailTime = format.parse(split[0]);
                                return orderDetailTime.getTime();
                            }
                        })
                ).map(new MapFunction<String, DetailOrder>() {
                    @Override
                    public DetailOrder map(String line) throws Exception {
                        // 2022-04-05 06:00:12,order_103,detail_1,milk,1,45.00
                        String[] array = line.split(",");
                        DetailOrder detailOrder = new DetailOrder();
                        detailOrder.setDetailTime(array[0]);
                        detailOrder.setOrderId(array[1]);
                        detailOrder.setDetailId(array[2]);
                        detailOrder.setGoodsName(array[3]);
                        detailOrder.setGoodsNumber(Integer.parseInt(array[4]));
                        detailOrder.setDetailMoney(Double.parseDouble(array[5]));
                        // 返回封装实体类对象
                        return detailOrder;
                    }
                });
        DataStream<DwdOrder> joinStream = orderStream.coGroup(orderDetailStream)
                .where(MainOrder::getOrderId).equalTo(DetailOrder::getOrderId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .apply(new CoGroupFunction<MainOrder, DetailOrder, DwdOrder>() {
                    /**
                     *对2个流窗口中匹配key的数据进行join操作，可以实现inner join和oute join（left、right和full）
                     * @param first 表示左边流中key相同的数据，此处指定的是订单表数据
                     * @param second    表示右边流中key相同的数据，此处指定的是订单详情表数据
                     * @param collector
                     * @throws Exception
                     */
                    @Override
                    public void coGroup(Iterable<MainOrder> first, Iterable<DetailOrder> second, Collector<DwdOrder> collector) throws Exception {
                        //  todo： 以左表为准，遍历数据
                        for (MainOrder mainOrder : first) {
                            DwdOrder dwdOrder = new DwdOrder();
                            dwdOrder.setOrderId(mainOrder.getOrderId());
                            dwdOrder.setOrderTime(mainOrder.getOrderTime());
                            dwdOrder.setUserId(mainOrder.getUserId());
                            dwdOrder.setAddress(mainOrder.getAddress());
                            dwdOrder.setOrderMoney(mainOrder.getOrderMoney());

                            // 定义变量，表示是否与右表关联
                            boolean isJoin = false;
                            // todo: 直接遍历右表数据，当且仅当右表中有数据时，才会执行遍历
                            for (DetailOrder detailOrder : second) {
                                isJoin = true;
                                // 如果有数据，就相当于关联，设置属性值
                                dwdOrder.setDetailId(detailOrder.getDetailId());
                                dwdOrder.setDetailOrderTime(detailOrder.getDetailTime());
                                dwdOrder.setDetailMoney(detailOrder.getDetailMoney());
                                dwdOrder.setGoodsName(detailOrder.getGoodsName());
                                dwdOrder.setGoodsNumber(detailOrder.getGoodsNumber());

                                collector.collect(dwdOrder);
                            }

                            if (!isJoin) {
                                // 如果右表没有阈值key相关的数据，说明没有关联成功，单独输出左右表数据，dodo：类似左外连接
                                collector.collect(dwdOrder);
                            }
                        }
                    }
                });
        // 4. 数据终端-sink
        joinStream.printToErr();
        // 5. 触发执行-execute
        env.execute("TumblingWindowCoGroupDemo");
    }
}